package com.org.eventcenter.eventcenter.service.impl;

import cn.miludeer.jsoncode.JsonCode;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.org.eventcenter.eventcenter.dao.DataFuseDao;
import com.org.eventcenter.eventcenter.dto.DataFuseDto;
import com.org.eventcenter.eventcenter.dto.Response;
import com.org.eventcenter.eventcenter.service.DataFuseService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author Evil Lee
 * @创建时间 2020/5/16
 * @描述
 */
@Slf4j
@Service
public class DataFuseServiceImpl implements DataFuseService {

    @Autowired
    DataFuseDao reqDao;


    @Autowired
    KafkaTemplate kafkaTemplate;

    @Resource
    RedisTemplate<String, List<DataFuseDto>> redisTemplate;

    /**
     * 数据融合
     *
     * @param data
     * @return
     */
    @Override
    public Response dataFuse(String data, String type) {
        List<DataFuseDto> reqs;
        //目标数据传输MAP
        JSONObject jsonObject;
        if (null != redisTemplate.opsForValue().get(type)) {
            log.info("我还存在着呢");
            reqs = redisTemplate.opsForValue().get(type);
            log.info("redis:{}", reqs);
        } else {
            log.info("我不存在了");
            reqs = reqDao.queryAll(type);
            redisTemplate.opsForValue().set(type, reqs);
        }
        if (null != reqs && !reqs.isEmpty()) {
            for (DataFuseDto req : reqs) {
                String template = req.getReqTemplate().toString();
                List<String> list = JSONObject.parseObject(req.getReqPram().toString(), List.class);
                for (String s : list) {
                    //属性组装
                    log.info("解析开始----------------data:{}", data);
                    if (JSONObject.parseObject(data).containsKey(s)) {
                        String value = JsonCode.getValue(data, "$." + s);
                        if (StringUtils.isNotBlank(value)) {
                            template = template.replace("$." + s, value);
                        }
                    } else {
                        log.info("我出去了");
                        break;
                    }

                }
                if (!template.contains("$.")) {
                    jsonObject = JSONObject.parseObject(template);
                    log.info("temp:{}", jsonObject);
                    if (0 == req.getIsNeedCache()) {
                        log.info("准备向场景引擎发送:{}", jsonObject);
                    } else {
                        log.info("往scenarioDispatcher发消息");
                        kafkaTemplate.send("scenario_dispatcher_topic", JSON.toJSONString(jsonObject));
                    }
                }
            }
        }
        return new Response();
    }

    /**
     * 事件中心业务数据整合接口上报
     *
     * @param object
     * @return
     */
    @Override
    public Response eventSupport(JSONObject object) {
        String type = "support";
        String s = JSON.toJSONString(object);
        return this.dataFuse(s, type);
    }
}
